import rx.Observable;
import rx.Subscriber;

public class M {
    public static void main(String[] args) {
        merge();
    }

    private static void merge() {

        //单层展平
        Observable<Observable<String>> observableObservable = Observable.create(new Observable.OnSubscribe<Observable<String>>() {


            public void call(Subscriber<? super Observable<String>> e) {
                for (int i = 8; i > 0; i--) {
                    e.onNext(Observable.<String>just("Item" + i));
                }
                e.onCompleted();
            }
        });
        Observable.merge(observableObservable).subscribe(new Subscriber<String>() {
            long start;

            @Override
            public void onStart() {
                start = System.nanoTime();
            }

            public void onNext(String value) {
                System.out.println(value);
            }

            public void onError(Throwable e) {

            }

            public void onCompleted() {
                System.out.println("time consumed:" + (System.nanoTime() - start));
            }
        });

        //三层展开
        Observable<Observable<Observable<String>>> observablesss = Observable.create(new Observable.OnSubscribe<Observable<Observable<String>>>() {

            public void call(Subscriber<? super Observable<Observable<String>>> e) {
                for (int i = 0; i < 8; i++) {
                    e.onNext(Observable.just(Observable.<String>just("Item== " + i)));
                }
                e.onCompleted();
            }
        });

        Observable<Observable<String>> merge = Observable.merge(observablesss);
        Observable.merge(merge).subscribe(new Subscriber<String>() {
            long start;

            @Override
            public void onStart() {
                start = System.nanoTime();
            }

            public void onNext(String value) {
                System.out.println(value);
            }

            public void onCompleted() {
                System.out.println("time consumed:" + (System.nanoTime() - start));
            }

            public void onError(Throwable e) {

            }
        });
    }
}
